package com.jml.mapreduce.reducejoin;

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

/**
 * 需求：
 * ● 有两个文件，一个是订单表，一个是商品表。
 *   ○ order.txt
 *     ■ id     pid   amount
 *     ■ 1001	 01	    1
 *     ■ 1002	 02	    2
 *     ■ 1003	 03	    3
 *     ■ 1004	 01	    4
 *     ■ 1005	 02	    5
 *     ■ 1006	 03	    6
 *   ○ pd.txt
 *     ■ pid    pname
 *     ■ 01	    小米
 *     ■ 02	    华为
 *     ■ 03	    格力
 * ● 将商品表的信息合并到订单表中：预期输出结果如下
 *      id	    pname	amount
 *      1001	小米	    1
 *      1004	小米	    4
 *      1002	华为	    2
 *      1005	华为	    5
 *      1003	格力	    3
 *      1006	格力	    6
 *
 * Map先把两个文件全部封装成OrderBean，没有的值就是空串或者0，不能是null
 * 默认是一个文件是一个Map处理
 *封装后
 *    id	pid    pname	amount
 *    1001	 01	     “”       1
 *    1002	 02	     “”       2
 *    1003	 03	     “”       3
 *    1004	 01	     “”       4
 *    1005	 02	     “”       5
 *    1006	 03	     “”       6
 *    0      01	     小米      0
 *    0      02	     华为      0
 *    0      03	     格力      0
 *
 * Map阶段：
 *      Map后compareTo排序，先按照pid排序，pid一致就按照pname排序，然后根据pid分组如下：
 *
 *    id	pid    pname	amount
 *    0      03	     格力      0
 *    1003	 03	     “”       3
 *    1006	 03	     “”       6
 *
 *    0      02	     华为      0
 *    1002	 02	     “”       2
 *    1005	 02	     “”       5
 *
 *    0      01	     小米      0
 *    1001	 01	     “”       1
 *    1004	 01	     “”       4
 *reduce阶段：
 *  分组后根据reduce解决输出
 *  先获取第一行数据，得到第一行数据的pname，然后遍历这一组剩余的数据，把pname进行赋值，再输出
 *
 *
 *
 */
public class OrderBean implements WritableComparable<OrderBean> {
    private String id;
    private String pid;
    private int amount;
    private String pname;

    /**
     * 按照pid排序, pid一样就把有pname的放在最前面(根据pname排序)
     */
    @Override
    public int compareTo(OrderBean o) {
        int i = this.pid.compareTo(o.pid);
        if (i != 0) {
            return i;
        } else {
            return o.pname.compareTo(this.pname);
        }
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(id);
        out.writeUTF(pid);
        out.writeInt(amount);
        out.writeUTF(pname);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.id = in.readUTF();
        this.pid = in.readUTF();
        this.amount = in.readInt();
        this.pname = in.readUTF();
    }

    @Override
    public String toString() {
        return id + "\t" + pname + "\t" + amount;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getPid() {
        return pid;
    }

    public void setPid(String pid) {
        this.pid = pid;
    }

    public int getAmount() {
        return amount;
    }

    public void setAmount(int amount) {
        this.amount = amount;
    }

    public String getPname() {
        return pname;
    }

    public void setPname(String pname) {
        this.pname = pname;
    }


}
